package com.haotian.kafka;

import java.io.IOException;
import java.net.UnknownHostException;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

/**
 * kafka 生产者 可 单利多线程调用
 * <p>
 * </p>
 * 
 * @author 张昊天 2018年5月16日
 */
public class _KafkaProducer extends KafkaBasic implements AutoCloseable {

	private KafkaProducer<String, String> producer;

	public _KafkaProducer(_KafkaConfig config) throws UnknownHostException, IOException {
		super(config);
		producer = KafkaFactory.createProducer(config);
	}

	/**
	 * 发送一个数据 topic 理解为队列 message 为消息 如果想马上发送 需要调用 flush 否则
	 * 将根据配置缓冲机制达到一定容量才统一发送提交
	 * <p>
	 * </p>
	 * 
	 * @param topic
	 * @param message
	 * @throws TimeoutException
	 * @throws ExecutionException
	 * @throws InterruptedException
	 */
	public RecordMetadata send(String topic, String message)
			throws InterruptedException, ExecutionException, TimeoutException {
		// 同步 返回 超时为5秒
		RecordMetadata recordMetadata = producer.send(new ProducerRecord<String, String>(topic, message)).get(5000,
				TimeUnit.MILLISECONDS);
		return recordMetadata;
	}

	/**
	 * 马上发送 当前缓冲中的信息
	 */
	public void flush() {
		producer.flush();
	}

	/**
	 * 获得生产者对象
	 * <p>
	 * </p>
	 * 
	 * @return
	 */
	public KafkaProducer<String, String> getProducer() {
		return this.producer;
	}

	/**
	 * 关闭生产者和 消费者
	 * <p>
	 * </p>
	 */
	public void close() {
		producer.close();
	}

}
